//! 消息路由器

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{info, warn, debug};

use nacos_core::{ClusterNode, NacosError};
use nacos_core::Result;
use nacos_core::traits::NetworkService;

use super::codec::{NetworkMessage, MessageType};

/// 路由规则
#[derive(Debug, Clone)]
pub struct RouteRule {
    pub source_pattern: String,
    pub target_pattern: String,
    pub protocol: String,
    pub priority: i32,
    pub enabled: bool,
}

impl RouteRule {
    pub fn new(
        source_pattern: String,
        target_pattern: String,
        protocol: String,
        priority: i32,
    ) -> Self {
        Self {
            source_pattern,
            target_pattern,
            protocol,
            priority,
            enabled: true,
        }
    }
}

/// 消息路由器
pub struct MessageRouter {
    rules: Arc<RwLock<HashMap<String, Vec<RouteRule>>>>,
    network_service: Arc<dyn NetworkService>,
    route_cache: Arc<RwLock<HashMap<String, String>>>,
}

impl MessageRouter {
    pub fn new(network_service: Arc<dyn NetworkService>) -> Self {
        Self {
            rules: Arc::new(RwLock::new(HashMap::new())),
            network_service,
            route_cache: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 添加路由规则
    pub async fn add_rule(&self,
        key: String,
        rule: RouteRule,
    ) {
        let mut rules = self.rules.write().await;
        rules.entry(key).or_insert_with(Vec::new).push(rule.clone());
        info!("添加路由规则: {:?}", rule);
    }

    /// 移除路由规则
    pub async fn remove_rule(&self,
        key: &str,
        priority: i32,
    ) {
        let mut rules = self.rules.write().await;
        if let Some(rule_list) = rules.get_mut(key) {
            rule_list.retain(|r| r.priority != priority);
            info!("移除路由规则: {}", key);
        }
    }

    /// 查找路由规则
    pub async fn find_route(
        &self,
        source: &str,
        target: &str,
        message_type: &MessageType,
    ) -> Option<RouteRule> {
        let cache_key = format!("{}:{}:{:?}", source, target, message_type);
        
        // 检查缓存
        {
            let cache = self.route_cache.read().await;
            if let Some(protocol) = cache.get(&cache_key) {
                let rules = self.rules.read().await;
                for rule_list in rules.values() {
                    for rule in rule_list {
                        if rule.protocol == *protocol && rule.enabled {
                            return Some(rule.clone());
                        }
                    }
                }
            }
        }

        // 查找匹配的规则
        let rules = self.rules.read().await;
        let mut matched_rules = Vec::new();

        for rule_list in rules.values() {
            for rule in rule_list {
                if !rule.enabled {
                    continue;
                }

                let source_match = self.matches_pattern(source, &rule.source_pattern);
                let target_match = self.matches_pattern(target, &rule.target_pattern);

                if source_match && target_match {
                    matched_rules.push(rule.clone());
                }
            }
        }

        // 按优先级排序并选择最高优先级的规则
        matched_rules.sort_by(|a, b| b.priority.cmp(&a.priority));
        
        let best_rule = matched_rules.first().cloned();
        
        // 缓存结果
        if let Some(ref rule) = best_rule {
            let mut cache = self.route_cache.write().await;
            cache.insert(cache_key, rule.protocol.clone());
        }

        best_rule
    }

    /// 模式匹配
    fn matches_pattern(&self,
        value: &str,
        pattern: &str,
    ) -> bool {
        if pattern == "*" {
            return true;
        }

        if pattern.ends_with('*') {
            let prefix = &pattern[..pattern.len() - 1];
            return value.starts_with(prefix);
        }

        if pattern.starts_with('*') {
            let suffix = &pattern[1..];
            return value.ends_with(suffix);
        }

        value == pattern
    }

    /// 路由消息
    pub async fn route_message(
        &self,
        message: NetworkMessage,
        _source_node: Option<ClusterNode>,
        _target_node: Option<ClusterNode>,
    ) -> Result<NetworkMessage> {
        let source = &message.header.source_node;
        let target = &message.header.target_node;
        
        let rule = self.find_route(source, target, &message.header.message_type).await;
        
        match rule {
            Some(rule) => {
                debug!("路由消息: {} -> {} via {}", source, target, rule.protocol);
                
                // 这里应该根据协议类型进行实际的路由
                let mut routed_message = message;
                routed_message.body.metadata.insert(
                    "protocol".to_string(),
                    rule.protocol.clone(),
                );
                
                Ok(routed_message)
            }
            None => {
                warn!("未找到路由规则: {} -> {}", source, target);
                Err(NacosError::Route("未找到路由规则".to_string()))
            }
        }
    }

    /// 获取所有规则
    pub async fn get_all_rules(&self,
    ) -> HashMap<String, Vec<RouteRule>> {
        self.rules.read().await.clone()
    }

    /// 清空路由缓存
    pub async fn clear_cache(&self,
    ) {
        let mut cache = self.route_cache.write().await;
        cache.clear();
        info!("路由缓存已清空");
    }
}

/// 负载均衡策略
#[derive(Debug, Clone, Copy)]
pub enum LoadBalanceStrategy {
    RoundRobin,
    Random,
    Weighted,
    LeastConnections,
}

/// 负载均衡器
pub struct LoadBalancer {
    strategy: LoadBalanceStrategy,
    weights: Arc<RwLock<HashMap<String, f64>>>,
    connections: Arc<RwLock<HashMap<String, usize>>>,
    round_robin_index: Arc<RwLock<HashMap<String, usize>>>,
}

impl LoadBalancer {
    pub fn new(strategy: LoadBalanceStrategy) -> Self {
        Self {
            strategy,
            weights: Arc::new(RwLock::new(HashMap::new())),
            connections: Arc::new(RwLock::new(HashMap::new())),
            round_robin_index: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 设置节点权重
    pub async fn set_weight(
        &self,
        node_id: String,
        weight: f64,
    ) {
        let mut weights = self.weights.write().await;
        weights.insert(node_id, weight);
    }

    /// 更新连接数
    pub async fn update_connections(
        &self,
        node_id: String,
        connections: usize,
    ) {
        let mut conn_map = self.connections.write().await;
        conn_map.insert(node_id, connections);
    }

    /// 选择目标节点
    pub async fn select_target(
        &self,
        targets: &[ClusterNode],
        key: &str,
    ) -> Option<ClusterNode> {
        if targets.is_empty() {
            return None;
        }

        match self.strategy {
            LoadBalanceStrategy::RoundRobin => {
                let mut index_map = self.round_robin_index.write().await;
                let index = index_map.entry(key.to_string()).or_insert(0);
                let selected = targets[*index % targets.len()].clone();
                *index += 1;
                Some(selected)
            }
            LoadBalanceStrategy::Random => {
                use rand::Rng;
                let mut rng = rand::thread_rng();
                let index = rng.gen_range(0..targets.len());
                Some(targets[index].clone())
            }
            LoadBalanceStrategy::Weighted => {
                let weights = self.weights.read().await;
                let total_weight: f64 = targets.iter()
                    .filter_map(|node| weights.get(&node.node_id))
                    .sum();
                
                if total_weight > 0.0 {
                    use rand::Rng;
                    let mut rng = rand::thread_rng();
                    let random = rng.gen_range(0.0..total_weight);
                    let mut current_weight = 0.0;
                    
                    for target in targets {
                        let weight = weights.get(&target.node_id).copied().unwrap_or(1.0);
                        current_weight += weight;
                        if random <= current_weight {
                            return Some(target.clone());
                        }
                    }
                }
                
                Some(targets[0].clone())
            }
            LoadBalanceStrategy::LeastConnections => {
                let connections = self.connections.read().await;
                targets.iter()
                    .min_by_key(|node| connections.get(&node.node_id).copied().unwrap_or(0))
                    .cloned()
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use nacos_core::{NodeState, ServiceInstance};

    #[tokio::test]
    async fn test_route_rule_matching() {
        let network_service = Arc::new(crate::service::MockNetworkService);
        let router = MessageRouter::new(network_service);

        let rule = RouteRule::new(
            "node-*".to_string(),
            "cluster-*".to_string(),
            "nacos_grpc".to_string(),
            1,
        );

        router.add_rule("test".to_string(), rule.clone()).await;

        let matched = router.find_route("node-1", "cluster-1", &MessageType::Request).await;
        assert!(matched.is_some());
        assert_eq!(matched.unwrap().protocol, "nacos_grpc");
    }

    #[tokio::test]
    async fn test_load_balancer_round_robin() {
        let balancer = LoadBalancer::new(LoadBalanceStrategy::RoundRobin);
        
        let targets = vec![
            ClusterNode {
                node_id: "node-1".to_string(),
                ip: "127.0.0.1".to_string(),
                port: 9001,
                state: NodeState::Running,
                ..Default::default()
            },
            ClusterNode {
                node_id: "node-2".to_string(),
                ip: "127.0.0.1".to_string(),
                port: 9002,
                state: NodeState::Running,
                ..Default::default()
            },
        ];

        let selected1 = balancer.select_target(&targets, "test").await.unwrap();
        let selected2 = balancer.select_target(&targets, "test").await.unwrap();
        
        assert_ne!(selected1.node_id, selected2.node_id);
    }

    #[tokio::test]
    async fn test_load_balancer_least_connections() {
        let balancer = LoadBalancer::new(LoadBalanceStrategy::LeastConnections);
        
        let targets = vec![
            ClusterNode {
                node_id: "node-1".to_string(),
                ip: "127.0.0.1".to_string(),
                port: 9001,
                state: NodeState::Running,
                ..Default::default()
            },
            ClusterNode {
                node_id: "node-2".to_string(),
                ip: "127.0.0.1".to_string(),
                port: 9002,
                state: NodeState::Running,
                ..Default::default()
            },
        ];

        balancer.update_connections("node-1".to_string(), 5).await;
        balancer.update_connections("node-2".to_string(), 1).await;

        let selected = balancer.select_target(&targets, "test").await.unwrap();
        assert_eq!(selected.node_id, "node-2");
    }
}

